package com.youxin.dataSet.api;

import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.ArrayList;
import java.util.List;

/**
 * 获取topn，
 * 注意，tuple不能使用scala中的，需要使用flink的Tuple2
 */
public class TopN {
    public static void main(String[] args) throws Exception {
        List<Tuple2<Integer,String>> list = new ArrayList<>();
        list.add(new Tuple2<>(1,"zhangsan"));
        list.add(new Tuple2<>(1,"lisi"));
        list.add(new Tuple2<>(1,"wangwu"));
        list.add(new Tuple2<>(2,"zhangsan"));
        list.add(new Tuple2<>(3,"zhangsan"));
        list.add(new Tuple2<>(3,"maliu"));
        list.add(new Tuple2<>(3,"tianqi"));
        list.add(new Tuple2<>(4,"xiaoba"));

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<Tuple2<Integer, String>> source = env.fromCollection(list);

        //打印三条数据
        source.first(3).print();
        System.out.println("========================");
        //分组，然后每组取出来两个数据
        source.groupBy(0).first(2).print();

        System.out.println("=======================");
        //分组topn
        source.groupBy(0).sortGroup(1, Order.DESCENDING).first(2).print();

        System.out.println("=======================");

        //全局排序   慎用（可以分组排序后进行全局排序）
        source.sortPartition(0,Order.ASCENDING).sortPartition(1,Order.DESCENDING).first(3).print();

    }
}
